package com.broadsoft.cluster.smartcluster.partition;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.jgroups.Address;
import org.jgroups.View;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
 

/**
 * 
 * FIXME: <A few lines of description of what class is about and what it does.>
 * 
 * <HR/>
 * <B>Modification Block</B>
 * <P>
 * Version: FIXME: 1.0.1 <BR>
 * Author: FIXME: luzhen Date: 2011-9-21 TAG: <A short name to find out all changed made by this modification.>
 * Comments: Created. Bug fix NO.: <If required>
 * </P>
 * <P>
 * ... Other Modifications following above structure.
 * </P>
 * 
 */
public class ClusterPartitionTable implements Serializable {

    private static final long serialVersionUID = 7585248941787420954L;
    private static final Logger logger = LoggerFactory.getLogger(ClusterPartitionTable.class);

    private static final AtomicInteger partitionIDGeneration = new AtomicInteger();
    private int totalPartitionNo = 1024; 
    private List<PartitionGroup> listPartitionGroup = new ArrayList<PartitionGroup>(); 
    private final transient ReadWriteLock partitionGroupLock = new ReentrantReadWriteLock();

    

    private final Map<Address, Worker> workerMap=null ;
    private final List<Worker> workers=null;

    
    public void addWorker(String name, Address address) {
        Worker newWorker = new Worker(name, address);
        partitionGroupLock.writeLock().lock();
        try {
            if (listPartitionGroup.size()==0) {
                init();
            }
            assignPartitionToNewWorker(name, address);
            
            workerMap.put(address, newWorker);
            workers.add(newWorker);
             
        } finally {
            partitionGroupLock.writeLock().unlock();
        }
    }
    
    private void assignPartitionToNewWorker(String name, Address address) {
        logger.info("Add new worker {} with address {}", name, address);
        int sizeOfWorker=workers.size();
        if(workers.isEmpty()){
            for(int i =0;i<totalPartitionNo;i++){
                listPartitionGroup.get(i).setMaster(address);
            }//end of for
        }else{
            for(int i =0;i<totalPartitionNo;i++){
                if(i%sizeOfWorker==0){
                    listPartitionGroup.get(i).setMaster(address);
                }
            }//end of for
        }
    }
    

    private void rebalanceSlavePartitions() {
        
    }
    
    /**
      * 
      * 
      */
    public void init() {
        partitionGroupLock.writeLock().lock();
        try {
            for (int i = 0; i < totalPartitionNo; i++) {
                PartitionGroup partitionGroup = new PartitionGroup(i);
                listPartitionGroup.add(partitionGroup);
            }
        } finally {
            partitionGroupLock.writeLock().unlock();
        }
    }

    public Address getSlaveAddressByPartitionID(int paritionID) {
        partitionGroupLock.readLock().lock();
        try {
            PartitionGroup partitionGroup = listPartitionGroup.get(paritionID);
            return partitionGroup.getSlave();
        } finally {
            partitionGroupLock.readLock().unlock();
        }
    }// end of getSlaveAddress

    public Address getMasterAddressByPartitionID(int paritionID) {
        partitionGroupLock.readLock().lock();
        try {
            PartitionGroup partitionGroup = listPartitionGroup.get(paritionID);
            return partitionGroup.getSlave();
        } finally {
            partitionGroupLock.readLock().unlock();
        }
    }// end of getMasterAddress

    /**
     * 
     * @param view
     * @return
     */
    public boolean enterPartition(View view) {
        Vector<Address> listOfAddress = view.getMembers();
        for (int i = 0; i < listOfAddress.size(); i++) {
            Address address = listOfAddress.get(i);
            
        }// end of for
        printDebugInfo();
        return true;
    }

    /**
     * 
     * FIXME: <A brief description of what the method does.> FIXME: Algorithm: <MUST exist if the method is more than 10
     * lines. Remove if it is simple.> <An Algorithm in pure English, describing the logic used in the implementation.
     * This should not contain to any Syntax of Java, in fact, it should be language independent, and should be targeted
     * to any new comer in the maintenance of the code.>
     * 
     * @param address
     * @return
     */
    public boolean leavePartition(Address address) {
        if (logger.isTraceEnabled()) {
            logger.trace("The node address[{}] leave from the cluster.", address);
        }
        return true;
    }

    private Integer generateParationId() {
        return partitionIDGeneration.incrementAndGet();
    }

 

    private void printDebugInfo() {
        StringBuilder sb = new StringBuilder();  
        sb.append("\n------Cluster  Partition Table Debug Info Begin------\n");
         
        sb.append("------Cluster  Partition Table Debug Info End------\n");
        if (logger.isTraceEnabled()) {
            logger.trace(sb.toString());
        }

    }

}

